package com.burges.net.assembly.cep

import akka.actor.FSM.Event
import org.apache.flink.cep.CEP
import org.apache.flink.cep.pattern.Pattern
import org.apache.flink.streaming.api.scala.{DataStream, OutputTag, StreamExecutionEnvironment}
import org.apache.flink.util.Collector

/**
  * 创建人    BurgessLee 
  * 创建时间   2020/2/26 
  * 描述     PatternAPI程序示例，
  */
object PatternAPIDEMO {

	def main(args: Array[String]): Unit = {
		/**
		  * 将温度大于35度的信号事件抽取出来
		  */
		val environment = StreamExecutionEnvironment.getExecutionEnvironment
		//创建输入事件流
		val inputStream: DataStream[Event] = ...
		//定义Pattern接口
		val pattern = Pattern.begin[Event]("start")
        		.where(_.getType == "temperature")
        		.next("middle")
        		.subtype(classOf[TempEvent])
        		.where(_.getType >= 35.0)
        		.followedBy("end")
        		.where(_.getName == "end")
		// 将创建好的Pattern应用在输入事件流上
		val patternStream = CEP.pattern(inputStream, pattern)
		// 获取事件触发结果
		// 调用select可以传入自定义的selectFunction，提取事件结果事件
		val rs: DataStream[Result] = patternStream.select(getResult(_))

		// Select抽取超时事件
		val patternStream = CEP.pattern(input, pattern)
		// 创建OutputTag并命名为timeout-output
		val timeoutTag = OutputTag[String]("timeout-output")
		//调用PatternStream select并指定timeoutTag
		patternStream.select(timeoutTag){
			//超时事件获取
			(pattern: Map[String, Iterable[Event]], timestamp: Long) => TimeoutEvent()//返回异常事件
		} {
			//正常事件获取
			(pattern: Map[String, Iterable[Event]]) => NormalEvent()//返回正常事件
		}
		// 调用getSideOutput方法，并指定timeoutTag将超时事件输出
		val  timeoutResult: DataStream[TimeoutEvent] = result.getSideOutput(tiemoutTag)


		// FlatSelect抽取超时事件
		val patternStream = CEP.pattern(input, pattern)
		// 创建OutputTag并命名为timeout-output
		val timeoutTag = OutputTag[String]("timeout-output")
		//调用PatternStream flatSelect并指定timeoutTag
		patternStream.flatSelect(timeoutTag){
			//超时事件获取
			(pattern: Map[String, Iterable[Event]], timestamp: Long, out: Collector[TimeoutEvent]) => out.collect(TimeoutEvent())//返回异常事件
		} {
			//正常事件获取
			(pattern: Map[String, Iterable[Event]], out: Collector[TimeoutEvent]) => out.collect(NormalEvent())//返回正常事件
		}
		// 调用getSideOutput方法，并指定timeoutTag将超时事件输出
		val  timeoutResult: DataStream[TimeoutEvent] = result.getSideOutput(tiemoutTag)
	}

	// 自定义的SelectFunction抽取正常事件
	def selectFunction(pattern : Map[String,  Iterable[IN]]) : OUT = {
		// 获取pattern中的startEvent
		val startEvent = pattern.get("start_pattern").get.next
		val middleEvent = pattern.get("middle_pattern").get.next
		OUT(startEvent, middleEvent)
	}

	// 自定义的flatSelectFn抽取正常事件
	def flatSelectFn(pattern : Map[String,  Iterable[IN]]): collector : Collector[OUT] ={
		// 获取pattern中的startEvent
		val startEvent = pattern.get("start_pattern").get.next
		val middleEvent = pattern.get("middle_pattern").get.next
		for (i <- 0 to startEvent.getValue) {
			collector.collect(OUT(startEvent, middleEvent))
		}
	}


}
